package com.flyfinger.client;

import com.flyfinger.codec.RpcDecoder;
import com.flyfinger.codec.RpcEncoder;
import com.flyfinger.codec.Serializer;
import com.flyfinger.codec.impl.JSONSerializer;
import com.flyfinger.handler.UserClientHandler;
import com.flyfinger.pojo.RpcRequest;
import com.flyfinger.pojo.RpcRequestUtil;
import com.flyfinger.pojo.ServiceLocation;
import com.flyfinger.register.CuratorClient;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * 消费者
 */
public class RPCConsumer {

	private final ExecutorService executorService =
			Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

	private final Serializer serializer = JSONSerializer.getInstance();

	private UserClientHandler userClientHandler;

	private CuratorFramework curatorClient;

	public RPCConsumer() {
		initZookeeper();
	}

	public void close() {
		curatorClient.close();
		if (userClientHandler != null) {
			userClientHandler.close();
		}
	}

	private void initNetty(String host, int port) throws Exception {
		if (userClientHandler != null) {
			userClientHandler.close();
		}
		userClientHandler = new UserClientHandler();
		Bootstrap bootstrap = new Bootstrap();
		bootstrap.group(new NioEventLoopGroup())
				.channel(NioSocketChannel.class)
				.option(ChannelOption.TCP_NODELAY, true)
				.handler(new ChannelInitializer<SocketChannel>() {
					protected void initChannel(SocketChannel socketChannel) throws Exception {
						ChannelPipeline pipeline = socketChannel.pipeline();
						pipeline.addFirst(new RpcDecoder(serializer));
						pipeline.addLast(new RpcEncoder(serializer));
						pipeline.addLast(userClientHandler);
					}
				});
		bootstrap.connect(host, port).sync();
	}

	@SuppressWarnings("unchecked")
	public <T> T createProxy(Class<? extends T> serviceClass) {
		return (T) Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
				new Class[]{serviceClass},
				(Object proxy, Method method, Object[] args) -> {
					final RpcRequest rpcRequest = RpcRequestUtil
							.initRpcRequestByMethodName(serviceClass, method.getName(), args);

					final ServiceLocation location = getServiceLocation();

					initNetty(location.getIp(), location.getPort());
					userClientHandler.setRpcRequest(rpcRequest);

					final Future<RpcRequest> submit = executorService.submit(userClientHandler);
					final RpcRequest request = submit.get();

					final long replyTime = System.currentTimeMillis() - request.getRequestId();
					location.setReplyTime(replyTime);

					return request.getResult();
				});
	}

	private void initZookeeper() {
		final CuratorFramework client = CuratorClient.curator();
		client.start();
		curatorClient = client;
	}

	private ServiceLocation getServiceLocation() throws Exception {
		final List<String> strings = getChildren(CuratorClient.ZK_CLIENT_PATH);

		final long currentTimeMillis = System.currentTimeMillis();

		long minReply = Integer.MAX_VALUE;
		ServiceLocation minServiceLocation = null;
		for (String path : strings) {
			final String[] zkPath = {CuratorClient.ZK_CLIENT_PATH, "/", path};
			final byte[] bytes = getData(zkPath);
			final ServiceLocation location = catchChildData(bytes);

			final long sec = getSec(currentTimeMillis, location.getReplyTime());
			if (sec >= 5) {
				location.setReplyTime(System.currentTimeMillis());
				setData(location, zkPath);
			}
			if (sec < minReply) {
				minReply = sec;
				minServiceLocation = location;
			}
		}
		if (minServiceLocation == null) {
			throw new RuntimeException("未找到服务器节点");
		}
		System.out.println(String.format("对服务器[%s]发起访问请求，响应时间为：[%s]秒", minServiceLocation.getPort(), minReply));
		return minServiceLocation;
	}

	private long getSec(long currentTimeMillis, long replyTime) {
		final long l = currentTimeMillis - replyTime;
		return Math.round((double) l / 1000);
	}

	private ServiceLocation catchChildData(byte[] data) {
		ServiceLocation location;
		try {
			location = JSONSerializer.getInstance().deserialize(ServiceLocation.class, data);
		} catch (IOException e) {
			throw new RuntimeException("解析zk删除节点数据失败");
		}
		if (location == null) {
			throw new RuntimeException("未获取到zk删除的节点数据");
		}
		return location;
	}

	public byte[] getData(String... path) throws Exception {
		return curatorClient.getData()
				.forPath(buildPath(path));
	}

	public Stat setData(ServiceLocation location, String... path) throws Exception {
		return curatorClient.setData()
				.forPath(buildPath(path),
						location.writeAsBytes());
	}

	public List<String> getChildren(String... path) throws Exception {
		return curatorClient
				.getChildren()
				.forPath(buildPath(path));
	}

	private String buildPath(String... param) {
		final StringBuilder builder = new StringBuilder();
		for (String s : param) {
			builder.append(s);
		}
		return builder.length() == 0 ? null : builder.toString();
	}


}
